Python for STEM Teachers
Oregon Curriculum Network
Lets talk about "context managers" in Python.
The "with context( ) as ctx:" construct provides a way of doing setup and teardown behind the scenes.
The context.__enter__ method creates the context, whereas context.__exit__ cleans up the environment as we exit the scope.
A context is typically used to establish a connection with an ecosystem resource, such as a database. The file type object returned by open( ) is also natively a context manager, self-closing upon __exit__.
Lets start with a SQL_DB class that establishes a connection to a database and stores the resulting connection and cursor objects at the class level. The Python DB API gives us the wherewithall to talk to any SQL engine, provided an appropriate driver has been imported.
The Python standard library hase SQLite natively, giving us an ideal tool with which to prototype a generic SQL engine. It may actually be enough for your needs in some cases.
In [1]:
import time
import sqlite3 as sql
import os
import sys
sys.path.append("/Users/kurner/Documents/classroom_labs")
class NoConnection(Exception):
pass
class SQL_DB: # a database
# class level parameters
backend = 'sqlite3'
user_initials = 'KTU'
timezone = int(time.strftime("%z", time.localtime()))
target_path = "/Users/kurner/Documents/classroom_labs" # current directory
db_name = os.path.join(target_path, 'glossary.db')
@staticmethod
def mod_date():
return time.mktime(time.gmtime()) # GMT time
@classmethod
def connect(cls):
try:
if cls.backend == 'sqlite3':
cls.conn = sql.connect(cls.db_name) # connection
cls.curs = cls.conn.cursor() # cursor
elif cls.backend == 'mysql': # not using this, gives idea
cls.conn = sql.connect(host='localhost',
user='root', port='8889')
cls.curs = cls.conn.cursor()
except Exception:
raise NoConnection
@classmethod
def disconnect(cls):
cls.conn.close()
So far, this class hasn't done any work. We'll use the context manager to make the actual connection, using this DB object.
Upon initialization, this context manager expects a database class as an argument. Context managers may be created without arguments as well. The __enter__ method is then immediately executed, and what it returns (maybe None) becomes the object named by the optional "as extension", in this case the object named dbx.
Dot notation takes us into the self of the DBcontext instance (dbx) which in turn anchors db, the database class, which was passed in to __init__. Inside the database class are the connection and cursor objects required by the DB API.
DBcontext has no specific knowledge about internal tables, simply connects and disconnects. We can subclass this generic class if wishing to add more database-specific methods. We'll start with a Glossary subclass, then move to Elements.
In [2]:
class DBcontext:
"""
Generic parent class for connecting and disconnecting
"""
def __init__(self, db):
self.db = db # references database class
def __enter__(self):
self.db.connect()
return self # <-- for use inside with scope
def __exit__(self, *stuff_happens):
self.db.disconnect()
if stuff_happens[0]:
print("Exception raised!")
print(stuff_happens)
return True # <-- if considered handled, otherwise False
return True
class Glossary(DBcontext):
"""
Subclass with custom methods for this particular database
"""
def create_table(self):
# https://www.sqlite.org/lang_droptable.html
self.db.curs.execute("""DROP TABLE IF EXISTS Glossary""")
self.db.curs.execute("""CREATE TABLE Glossary
(gl_term text PRIMARY KEY,
gl_definition text,
updated_at int,
updated_by text)""")
def save_term(self, *the_data):
query = ("INSERT INTO Glossary "
"(gl_term, gl_definition, updated_at, updated_by) "
"VALUES ('{}', '{}', {}, '{}')".format(*the_data))
# print(query)
self.db.curs.execute(query)
self.db.conn.commit()
In the "with scope" below, we create a new table by calling a method saved inside the context manager itself. The a text file is read in and each line gets saved to a row in the Glossary table. Again, the context manager has a method for inserting rows.
In [3]:
with Glossary(SQL_DB) as dbx: # <--- dbx returned by __enter__
# for testing __exit__ in case of an exception
# raise NoConnection
dbx.create_table()
FILE = os.path.join(dbx.db.target_path, "glossary.txt")
with open(FILE, 'r', encoding='UTF-8') as gloss:
lines = gloss.readlines()
for line in lines:
if len(line.strip()) == 0:
continue
term, definition = line.split(":", 1)
right_now = dbx.db.mod_date()
dbx.save_term(term[2:].strip(), definition.strip(), right_now, dbx.db.user_initials)
We now have a resuable context that will connect and disconnect from our database.
Lets use it to query the Glossary table...
In [4]:
with Glossary(SQL_DB) as dbx:
query = "SELECT gl_term, gl_definition FROM Glossary ORDER BY gl_term"
dbx.db.curs.execute(query) # gets the data
print("{:^80}".format("GLOSSARY OF TERMS"))
print("-" * 80)
print("Term |Abbreviated Definition " )
print("-" * 80)
for term in dbx.db.curs.fetchmany(10): # fetchone(), fetchmany(n), fetchall()
print("{:35} | {:45}".format(term[0], term[1][:45]))
OK, that was fun. We were able to obtain a partial listing of what's in the Glossary table. Connecting and disconnecting was handled behind the scenes.
Now lets subclass DBcontext once again and add methods specific to a Periodic Table of the Elements.
In [5]:
import chem_stuff
# modify database class to point to a different database file
SQL_DB.db_name = os.path.join(SQL_DB.target_path, 'periodic_table.db')
class ChemContext(DBcontext):
"""
Subclass with custom methods for this particular database
"""
def create_table(self):
# https://www.sqlite.org/lang_droptable.html
self.db.curs.execute("""DROP TABLE IF EXISTS Elements""")
self.db.curs.execute("""CREATE TABLE Elements
(elem_protons int PRIMARY KEY,
elem_symbol text,
elem_long_name text,
elem_mass float,
elem_series text,
updated_at int,
updated_by text)""")
def save_term(self, *the_data):
query = ("INSERT INTO Elements "
"(elem_protons, elem_symbol, elem_long_name, elem_mass, elem_series,"
"updated_at, updated_by) "
"VALUES ({}, '{}', '{}', {}, '{}', {}, '{}')".format(*the_data))
# print(query)
self.db.curs.execute(query)
self.db.conn.commit()
with ChemContext(SQL_DB) as dbx:
dbx.create_table()
FILE = os.path.join(dbx.db.target_path, "periodic_table.json")
chem_stuff.load_elements(FILE) # uses imported module to read JSON
for atom in chem_stuff.all_elements.values():
right_now = dbx.db.mod_date()
dbx.save_term(atom.protons, atom.symbol, atom.long_name, atom.mass, atom.series,
right_now, dbx.db.user_initials)
To test that our data table has indeed been populated, lets connect again and run a select query. We'll be able to print the rows directly. We're not bothering with an Atom class this time, just publishing row data directly from tuples.
In [6]:
with DBcontext(SQL_DB) as dbx: # <--- dbx returned by __enter__
query = ("SELECT elem_symbol, elem_long_name, elem_protons, elem_mass, elem_series"
" FROM Elements ORDER BY elem_protons")
dbx.db.curs.execute(query)
print("{:^70}".format("PERIODIC TABLE OF THE ELEMENTS"))
print("-" * 70)
print("Symbol |Long Name |Protons |Mass |Series " )
print("-" * 70)
for the_atom in dbx.db.curs.fetchall():
print("{:6} | {:20} | {:6} | {:5.2f} | {:15}".format(the_atom[0],
the_atom[1],
the_atom[2],
the_atom[3],
the_atom[4]))